package cn.galudisu.reactivemanifesto

import java.time.ZonedDateTime
import java.time.format.DateTimeFormatter

import akka.actor.{Actor, PoisonPill}
import akka.event.{Logging, LoggingAdapter}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model._
import akka.pattern.{CircuitBreaker, CircuitBreakerOpenException}
import akka.stream.ActorMaterializer
import cn.galudisu.reactivemanifesto.event._

import scala.concurrent.{Future, TimeoutException}
import scala.concurrent.duration._
import scala.language.implicitConversions

/**
  * 熔断器
  */
trait BasicCircuitBreaker extends Actor {
  import GospelConfig._

  import context.{dispatcher, system}

  implicit val mat: ActorMaterializer
  val eventBus: CircuitBreakerBus

  lazy val logger: LoggingAdapter = Logging.getLogger(context.system, this)
  val nows: String = ZonedDateTime
                     .now()
                     .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssZ"))

  def open: Unit = logger.debug("circuit breaker open at {}", nows)
  def close: Unit = logger.debug("circuit breaker closed at {}", nows)
  def halfOpen: Unit = logger.debug("circuit breaker half-open at {}", nows)

  def uris: Set[Uri]

  def breaker: CircuitBreaker =
    new CircuitBreaker(
      context.system.scheduler,
      maxFailures = max_failures,
      callTimeout = call_timeout.seconds,
      resetTimeout = reset_timeout.seconds
    ).onOpen(open)
     .onClose(close)
     .onHalfOpen(halfOpen)

  protected def fuse(uri: Uri): Future[HttpResponse] =
    Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = uri))

  def invoke: Unit =
    uris.foreach { uri =>
      breaker.withCircuitBreaker(fuse(uri)).map { resp =>
        resp.entity.discardBytes()
        resp.status match {
          case Informational(status) =>
            logger.debug("service available, 10x series")
            eventBus.publish(Succ(status, uri, None))
            suicide

          case Success(status) =>
            logger.debug("service available, 20x series")
            eventBus.publish(Succ(status, uri, None))
            suicide

          case Redirection(status) =>
            logger.debug("service available, 30x series")
            eventBus.publish(Succ(status, uri, None))
            suicide

          case ClientError(_) | ServerError(_) =>
            logger.debug("service failure, 40x and 50x series relate {}", uri)
            eventBus.publish(Err(resp.status.intValue(), uri, None))
            suicide

          case _ =>
            logger.debug("internal server error occur, crash crash!!!")
            eventBus.publish(Err(resp.status.intValue(), uri, None))
            suicide
        }
      }.recover {
        case e: CircuitBreakerOpenException =>
          logger.debug("circuit breaker open exception {}", e.getMessage)
          eventBus.publish(Rec(500, uri, Some(e.remainingDuration)))
          suicide

        case e: TimeoutException =>
          logger.debug("illustrate timeout error, {} careful! all the api down {}",Console.RED, Console.RESET)
          eventBus.publish(Tot(500, uri, Some(call_timeout.seconds)))
          suicide

        case e@_ =>
          logger.debug("service no response {}", e)
          eventBus.publish(Err(500, uri, Some(call_timeout.seconds)))
          suicide
      }
    }

  private def suicide = self ! PoisonPill
}